package com.ichoice.rabbitmq.rpc.helper;

import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.ichoice.rabbitmq.rpc.helper.consts.CommonContant;
import com.ichoice.rabbitmq.rpc.helper.model.CallBackModel;
import com.ichoice.rabbitmq.rpc.helper.model.ProducerModel;
import com.ichoice.rabbitmq.rpc.helper.service.Consumer;
import com.ichoice.rabbitmq.rpc.helper.util.JsonUtil;
import com.ichoice.rabbitmq.rpc.helper.util.MqMetadataUtils;
import com.ichoice.rabbitmq.rpc.helper.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

/**
 * @author xianrensawang
 * @date 2021/3/26
 */
@Slf4j
public class ExecuteBlinding {

    private final RpcRabbitContextAware rpcRabbitContextAware;

    private final Consumer consumers;

    public ExecuteBlinding(Consumer consumers) {
        this.rpcRabbitContextAware = (RpcRabbitContextAware) SpringUtil.getBean(RpcRabbit.class);
        this.consumers = consumers;
    }

    /**
     * 初始化绑定
     *
     * @param appName appName
     */
    public void init(String appName) {
        // 获取监听类
        SimpleMessageListenerContainer container = SpringUtil.getBean("messageListenerContainer");
        // 消息回传队列
        String backQueue = CommonContant.backQueue;
        // 当前服务的queue,用于存储其他服务过来的请求
        String selfQueue = MqMetadataUtils.selfQueue(appName);
        // 延迟队列
        String delayQueue = MqMetadataUtils.selfDelayQueue(appName);
        container.addQueueNames(backQueue, selfQueue, delayQueue);
        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            MessageProperties properties = message.getMessageProperties();
            String consumerQueue = properties.getConsumerQueue();
            byte[] body = message.getBody();
            // 获取回传queue
            try{
                if (backQueue.equals(consumerQueue)) {
                    clientTask(body);
                    // 接收其他服务的请求
                } else if (selfQueue.equals(consumerQueue) || delayQueue.equals(consumerQueue)) {
                    serverTask(body);
                }
            }catch (Exception e){
                log.error("mq消费失败:{}",e.getMessage());
//                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
////                throw e; // 这里抛出异常,不ack
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        });
    }


    /**
     * 消费其他服务端请求
     *
     * @param body 请求体
     */
    private void serverTask(byte[] body) throws Exception {
        try {
            JSONObject jsonObject = JSONUtil.parseObj(new String(body, StandardCharsets.UTF_8), JSONConfig.create().setIgnoreNullValue(true));
            ProducerModel producerModel = JSONUtil.toBean(jsonObject, ProducerModel.class);
            //ProducerModel producerModel = JsonUtil.toObject(new String(body, StandardCharsets.UTF_8), ProducerModel.class);
            if(Objects.nonNull(producerModel)){
                consumers.consumer(producerModel);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            throw e;
        }
    }


    /**
     * 消费其他服务回参
     *
     * @param body 请求体
     */
    private void clientTask(byte[] body) throws Exception {
        try {
            JSONObject jsonObject = JSONUtil.parseObj(new String(body, StandardCharsets.UTF_8), JSONConfig.create().setIgnoreNullValue(true));
            CallBackModel callBackModel = JSONUtil.toBean(jsonObject, CallBackModel.class);
            //CallBackModel callBackModel = JsonUtil.toObject(new String(body, StandardCharsets.UTF_8), CallBackModel.class);
            if(Objects.nonNull(callBackModel)){
                rpcRabbitContextAware.consumer(callBackModel);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            throw e;
        }
    }
}
